#pragma once

#include "config.h"

#if USE_AWS_S3

#include <Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.h>
#include <memory>
#include <IO/S3/S3Capabilities.h>
#include <IO/S3Settings.h>
#include <Common/MultiVersion.h>
#include <Common/ObjectStorageKeyGenerator.h>


namespace DB
{

namespace S3RequestSetting
{
    extern const S3RequestSettingsBool read_only;
}


class S3ObjectStorage : public IObjectStorage
{
private:
    friend class S3PlainObjectStorage;

    S3ObjectStorage(
        const char * logger_name,
        std::unique_ptr<S3::Client> && client_,
        std::unique_ptr<S3Settings> && s3_settings_,
        S3::URI uri_,
        const S3Capabilities & s3_capabilities_,
        ObjectStorageKeyGeneratorPtr key_generator_,
        const String & disk_name_,
        bool for_disk_s3_ = true)
        : uri(uri_)
        , disk_name(disk_name_)
        , client(std::move(client_))
        , s3_settings(std::move(s3_settings_))
        , s3_capabilities(s3_capabilities_)
        , key_generator(std::move(key_generator_))
        , log(getLogger(logger_name))
        , for_disk_s3(for_disk_s3_)
    {
    }

public:
    template <typename... Args>
    explicit S3ObjectStorage(std::unique_ptr<S3::Client> && client_, Args && ...args)
        : S3ObjectStorage("S3ObjectStorage", std::move(client_), std::forward<Args>(args)...)
    {
    }

    std::string getName() const override { return "S3ObjectStorage"; }

    std::string getCommonKeyPrefix() const override { return uri.key; }

    std::string getDescription() const override { return uri.endpoint; }

    ObjectStorageType getType() const override { return ObjectStorageType::S3; }

    bool exists(const StoredObject & object) const override;

    std::unique_ptr<ReadBufferFromFileBase> readObject( /// NOLINT
        const StoredObject & object,
        const ReadSettings & read_settings,
        std::optional<size_t> read_hint = {}) const override;

    SmallObjectDataWithMetadata readSmallObjectAndGetObjectMetadata( /// NOLINT
        const StoredObject & object,
        const ReadSettings & read_settings,
        size_t max_size_bytes,
        std::optional<size_t> read_hint = {}) const override;

    /// Open the file for write and return WriteBufferFromFileBase object.
    std::unique_ptr<WriteBufferFromFileBase> writeObject( /// NOLINT
        const StoredObject & object,
        WriteMode mode,
        std::optional<ObjectAttributes> attributes = {},
        size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE,
        const WriteSettings & write_settings = {}) override;

    void listObjects(const std::string & path, RelativePathsWithMetadata & children, size_t max_keys) const override;

    ObjectStorageIteratorPtr iterate(const std::string & path_prefix, size_t max_keys, bool with_tags) const override;

    /// Uses `DeleteObjectRequest`.
    void removeObjectIfExists(const StoredObject & object) override;

    /// Uses `DeleteObjectsRequest` if it is allowed by `s3_capabilities`, otherwise `DeleteObjectRequest`.
    /// `DeleteObjectsRequest` does not exist on GCS, see https://issuetracker.google.com/issues/162653700 .
    void removeObjectsIfExist(const StoredObjects & objects) override;

    void tagObjects(const StoredObjects & objects, const std::string & tag_key, const std::string & tag_value) override;

    ObjectMetadata getObjectMetadata(const std::string & path, bool with_tags) const override;

    std::optional<ObjectMetadata> tryGetObjectMetadata(const std::string & path, bool with_tags) const override;

    void copyObject( /// NOLINT
        const StoredObject & object_from,
        const StoredObject & object_to,
        const ReadSettings & read_settings,
        const WriteSettings & write_settings,
        std::optional<ObjectAttributes> object_to_attributes = {}) override;

    void copyObjectToAnotherObjectStorage( /// NOLINT
        const StoredObject & object_from,
        const StoredObject & object_to,
        const ReadSettings & read_settings,
        const WriteSettings & write_settings,
        IObjectStorage & object_storage_to,
        std::optional<ObjectAttributes> object_to_attributes = {}) override;

    void shutdown() override;

    void startup() override;

    void applyNewSettings(
        const Poco::Util::AbstractConfiguration & config,
        const std::string & config_prefix,
        ContextPtr context,
        const ApplyNewSettingsOptions & options) override;

    std::string getObjectsNamespace() const override { return uri.bucket; }

    bool isRemote() const override { return true; }

    bool supportParallelWrite() const override { return true; }

    ObjectStorageKeyGeneratorPtr createKeyGenerator() const override;

    bool isReadOnly() const override { return s3_settings.get()->request_settings[S3RequestSetting::read_only]; }

    std::shared_ptr<const S3::Client> getS3StorageClient() override;
    std::shared_ptr<const S3::Client> tryGetS3StorageClient() override;

    S3::URI getURI() const { return uri; }
    S3Settings getS3Settings() const { return *s3_settings.get(); }
private:
    void removeObjectImpl(const StoredObject & object, bool if_exists);
    void removeObjectsImpl(const StoredObjects & objects, bool if_exists);

    const S3::URI uri;

    std::string disk_name;

    MultiVersion<S3::Client> client;
    MultiVersion<S3Settings> s3_settings;
    S3Capabilities s3_capabilities;

    const ObjectStorageKeyGeneratorPtr key_generator;

    LoggerPtr log;

    const bool for_disk_s3;
};

}

#endif
